// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: jhkim

#pragma once

#include <map>
#include <string>

#include <boost/date_time/posix_time/posix_time.hpp>
#include <boost/stacktrace.hpp>
#include <simdjson.h>

#include "coin/proto/coin_market_enums.pb.h"
#include "coin/proto/coin_order_enums.pb.h"
#include "coin2/base/conversion.h"
#include "coin2/base/time.h"
#include "coin2/exchange/base/feed/update.h"
#include "coin2/exchange/base/feed_util/book_builder.h"
#include "coin2/exchange/base/feed_util/trade.h"
#include "coin2/exchange/base/feed_util/kline.h"
#include "coin2/exchange/base/market/enums.h"

namespace coin2::exchange::binance_base::feed_v1 {
namespace impl {
using coin2::exchange::base::market::ExchangeType;
using coin2::exchange::base::market::MarketType;
using coin2::exchange::base::feed::FeedUpdateManager;
using coin2::exchange::base::feed_util::Index;
using namespace coin::proto;

/**
 * bookTicker (
 *  u: order book updateid,
 *  s: symbol,
 *  b: bidp, B: bidq
 *  a: askp, A: askq
 *
 * aggTrade (
 *  e: event type ["aggTrade"]
 *  E: event Time
 *  s: symbol ["BTCUSDT"]
 *  a: aggregated trade ID,
 *  p: price
 *  q: qty
 *  f: first trade ID
 *  l: last trade id
 *  T: trade time
 *  m: is the buyer the market maker? [true/false]
 * )
 */

inline coin::proto::OrderSide GetOrderSide(const std::string& side) {
  if (side == "BUY") {
    return coin::proto::OrderSide::BUY_ORDER;
  } else if (side == "SELL") {
    return coin::proto::OrderSide::SELL_ORDER;
  } else {
    return coin::proto::OrderSide::UNKNOWN_ORDER_SIDE;
  }
}

inline coin::proto::OrderSide GetOrderSide(std::string_view side) {
  if (side == "BUY") {
    return coin::proto::OrderSide::BUY_ORDER;
  } else if (side == "SELL") {
    return coin::proto::OrderSide::SELL_ORDER;
  } else {
    return coin::proto::OrderSide::UNKNOWN_ORDER_SIDE;
  }
}


inline coin::proto::OrderType GetOrderType(const std::string& type) {
  if (type == "LIMIT") {
    return coin::proto::OrderType::LIMIT_ORDER;
  } else if (type == "MARKET") {
    return coin::proto::OrderType::MARKET_ORDER;
  } else {
    return coin::proto::OrderType::UNKNOWN_ORDER_TYPE;
  }
}

inline coin::proto::OrderType GetOrderType(std::string_view type) {
  if (type == "LIMIT") {
    return coin::proto::OrderType::LIMIT_ORDER;
  } else if (type == "MARKET") {
    return coin::proto::OrderType::MARKET_ORDER;
  } else {
    return coin::proto::OrderType::UNKNOWN_ORDER_TYPE;
  }
}

inline coin::proto::OrderState GetOrderState(const std::string& state) {
  if (state == "PENDING_CANCEL" || state == "REJECTED" ||
      state == "EXPIRED" || state == "CANCELED" || state == "FILLED") {
    return coin::proto::OrderState::DEAD_ORDER;
  } else if (state == "NEW" || state == "CONFIRMED" || state == "PARTIALLY_FILLED") {
    return coin::proto::OrderState::WORKING_ORDER;
  } else {
    throw std::runtime_error("Unknown order status! " + state);
  }
}

inline coin::proto::OrderState GetOrderState(std::string_view state) {
  if (state == "PENDING_CANCEL" || state == "REJECTED" ||
      state == "EXPIRED" || state == "CANCELED" || state == "FILLED") {
    return coin::proto::OrderState::DEAD_ORDER;
  } else if (state == "NEW" || state == "CONFIRMED" || state == "PARTIALLY_FILLED") {
    return coin::proto::OrderState::WORKING_ORDER;
  } else {
    throw std::runtime_error("Unknown order status! " + std::string(state));
  }
}

inline coin::proto::OrderDuration GetOrderDuration(const std::string& duration) {
  if (duration == "IOC") {
    return coin::proto::OrderDuration::IOC_ORDER;
  } else if (duration == "GTC") {
    return coin::proto::OrderDuration::GTC_ORDER;
  } else {
    throw std::runtime_error("Unknown order duration! " + duration);
  }
}

inline coin::proto::OrderDuration GetOrderDuration(std::string_view duration) {
  if (duration == "IOC") {
    return coin::proto::OrderDuration::IOC_ORDER;
  } else if (duration == "GTC") {
    return coin::proto::OrderDuration::GTC_ORDER;
  } else {
    throw std::runtime_error("Unknown order duration! " + std::string(duration));
  }
}

class BinanceBaseFeedUpdateParser {
 public:
  template<class T>
  static void ParseTimestamp(
      base::executor::LifecycleTimestamp* ts,
      T&&  msg) {
    if (int64_t v; !msg["T"].get_int64().get(v)) {
      ts->set_exchange_transact_timestamp(v * 1'000'000LL);
    }
    if (int64_t v; !msg["E"].get_int64().get(v)) {
      ts->set_exchange_publish_timestamp(v * 1'000'000LL);
    }
    // if (msg.HasMember("u")) {
    //   ts->set_exchange_publish_sequence(int64_t(msg["u"]));
    // }
  }

  //////////// L1 Realtime Book
  // e.g.{"stream":"btcusdt@bookTicker","data":{"u":3227874550,"s":"BTCUSDT","b":"8414.26","B":"3.242","a":"8414.62","A":"0.544"}}
  template<class T>
  static bool ParseL1RealtimeBookItem(
      BookBuilder* book,
      T&&  msg,
      std::string_view native_symbol,
      int64_t packet_timestamp) {
    int64_t exchange_timestamp;
    // will always have this field for futures.
    if (int64_t v;!msg["T"].get_int64().get(v)) {
      exchange_timestamp = v * 1'000'000LL;
    } else if (!msg["E"].get_int64().get(v)) {
      exchange_timestamp = v * 1'000'000LL;
    } else {
      exchange_timestamp = 0L;
    }
    /*
    2020-05-15

    New fields in payloads of <symbol>@bookTicker and !bookTicker:
    E for event time
    T for transactiontime

    */
    if (exchange_timestamp < book->Timestamp()) {
      SPDLOG_WARN_EVERY_NS(
          5'000'000'000L,
          exchange_timestamp,
          "{} {} {} {}, {}, {}",
          "Book of",
          native_symbol,
          "ET timestamp inverse!",
          exchange_timestamp,
          book->Timestamp(),
          packet_timestamp);
      return false;
    }
    auto tx = book->NewTransaction(exchange_timestamp);
    tx.Clear();
    tx.UpdateAsk(double(msg["a"].get_double_in_string()), double(msg["A"].get_double_in_string()));
    tx.UpdateBid(double(msg["b"].get_double_in_string()), double(msg["B"].get_double_in_string()));
    return true;
  }

  //////////// Diff & Snapshot Book
  // all book events use "a" and "b"
  // except spot+snapshot use "asks" and "bids"
  template<class T>
  static bool ParseBook(
      BookBuilder* book,
      T&&  msg,
      MarketType market,
      BookBuilder::BookOption book_type,
      std::string_view native_symbol,
      int64_t packet_timestamp) {
    if (market == MarketType::Spot && book_type == BookBuilder::SNAPSHOT) {
      auto tx = book->NewTransaction(0);  // Binance spot snapshot has no exchange-timestamp
      tx.Clear();
      double result[2] = {};
      for (auto && pq : simdjson::ondemand::array(msg["asks"])) {
        int i = 0;
        for (auto && p : pq)
        {
          if (i > 1) break;
          result[i++] = double(p.get_double_in_string());
        }
        
        tx.UpdateAsk(result[0], result[1]);
      }
      for (auto && pq : simdjson::ondemand::array(msg["bids"])) {
        int i = 0;
        for (auto && p : pq)
        {
          if (i > 1) break;
          result[i++] = double(p.get_double_in_string());
        }
        tx.UpdateBid(result[0], result[1]);
      }
      return true;
    } else {
      int64_t exchange_timestamp = 0L;
      // E: exchange sent time
      // T: actual occurrence of trade or book
      if (!msg["T"].error()) {
        exchange_timestamp = int64_t(msg["T"]) * 1'000'000LL;
      } else {
        exchange_timestamp = int64_t(msg["E"]) * 1'000'000LL;
      }
      if (exchange_timestamp < book->Timestamp()) {
        SPDLOG_WARN_EVERY_NS(
            5'000'000'000L,
            exchange_timestamp,
            "{} {} {} {}, {}, {}",
            "Book of",
            native_symbol,
            "ET timestamp inverse!",
            exchange_timestamp,
            book->Timestamp(),
            packet_timestamp);
        return false;
      }

      auto tx = book->NewTransaction(exchange_timestamp);
      if (book_type == BookBuilder::SNAPSHOT) {
        tx.Clear();
      } else if (book_type == BookBuilder::LIGHT_BBO) {
        tx.BboClear();
      }
      double result[2] = {};
      for (auto && pq : simdjson::ondemand::array(msg["a"])) {
        int i = 0;
        for (auto && p : pq)
        {
          if (i > 1) break;
          result[i++] = double(p.get_double_in_string());
        }
        
        tx.UpdateAsk(result[0], result[1]);
      }
      for (auto && pq : simdjson::ondemand::array(msg["b"])) {
        int i = 0;
        for (auto && p : pq)
        {
          if (i > 1) break;
          result[i++] = double(p.get_double_in_string());
        }
        tx.UpdateBid(result[0], result[1]);
      }
      return true;
    }
  }

  //////////// Trade
  template<class T>
  static void ParseTradeItem(
      T&& msg,
      Trade* trade,
      int64_t timestamp) {
    CHECK(trade);
    // this should be T: trade time.
    trade->timestamp = timestamp * 1'000'000LL;
    trade->price = double(msg["p"].get_double_in_string());//(std::string(msg["p"])); 
    trade->fill_qty = double(msg["q"].get_double_in_string());

    // is the buyer the market maker?
    
    trade->side = bool(msg["m"]) ? coin2::exchange::feed::TRADE_SELL_SIDE
                                 : coin2::exchange::feed::TRADE_BUY_SIDE;
  }

  //////////// Liquidation Order
  template<class T>
  static void ParseLiquidationOrder(
      T &&  msg,
      LiquidationOrder* liquidation_order,
      const int64_t timestamp) {
    CHECK(liquidation_order);
    liquidation_order-> Clear();
    liquidation_order->timestamp = timestamp * 1'000'000LL;
    liquidation_order->price = double(msg["p"].get_double_in_string());
    liquidation_order->qty = double(msg["q"].get_double_in_string());
    liquidation_order->avg_price = double(msg["ap"].get_double_in_string());
    liquidation_order->fill_qty = double(msg["z"].get_double_in_string());

    liquidation_order->order_side = GetOrderSide(msg["S"].get_string());
    liquidation_order->order_type = GetOrderType(msg["o"].get_string());
    liquidation_order->order_duration = GetOrderDuration(msg["f"].get_string());
    liquidation_order->order_state = GetOrderState(msg["X"].get_string());
  }

  template<class T>
  static void ParseOpenInterest(
      T && msg,
      OpenInterest* open_interest) {
    open_interest->timestamp = int64_t(msg["time"]) * 1'000'000LL;
    open_interest->open_interest_qty = double(msg["openInterest"].get_double_in_string());
  }

  //////////// Mark Price
  template<class T>
  static void ParseMarkPrice(
      T&&  msg,
      MarkPrice* mark_price,
      int64_t timestamp
      ) {
    CHECK(mark_price);
    mark_price->Clear();
    mark_price->timestamp = timestamp * 1'000'000LL;
    mark_price->mark_price = double(msg["p"].get_double_in_string());
  }

  //////////// Index
  template<class T>
  static void ParseIndex(
      T&& msg,
      Index* index,
      int64_t timestamp
      ) {
    CHECK(index);
    index->Clear();
    index->timestamp = timestamp * 1'000'000LL;
    index->price = double(msg["i"].get_double_in_string());
  }

  //////////// Index
  template<class T>
  static void ParseIndex(
      T&&  msg,
      Index* index,
      std::string* underlying_spot,
      int64_t timestamp
      ) {
    CHECK(index);
    index->Clear();
    index->timestamp = timestamp * 1'000'000LL;
    index->price = double(msg["p"].get_double_in_string());
    std::string_view str = msg["i"].get_string();
    *underlying_spot = std::string(str.begin(), str.end());
  }

  //////////// Funding Rate
  template<class T>
  static void ParseFundingRate(
      T&&  msg,
      FundingRate* funding_rate,
      int64_t timestamp) {
    CHECK(funding_rate);
    funding_rate->Clear();
    funding_rate->timestamp = timestamp * 1'000'000LL;
    funding_rate->funding_rate = double(msg["r"].get_double_in_string());
    funding_rate->funding_time = int64_t(msg["T"]) * 1'000'000LL;
  }

  //////////// Nav Rate
  template<class T>
  static void ParseNav(
      T&&  msg,
      Nav* nav) {
    CHECK(nav);
    nav->Clear();
    nav->timestamp = int64_t(msg["E"]) * 1'000'000LL;
    nav->actual_leverage = double(msg["l"]);
    nav->nav = double(msg["n"]);
    nav->outstanding = double(msg["m"]);

    for (auto && ua : simdjson::ondemand::array(msg["b"])) {
      std::string_view str = ua["s"].get_string();
      nav->basket.emplace_back(::std::string(str.begin(), str.end()), double(ua["n"]));
    }
  }

  static fastfeed::proto::KlineInterval ParseKlineInterval(std::string_view interval) {
    static std::map<std::string, fastfeed::proto::KlineInterval> interval_map = {
      {"1m", fastfeed::proto::KLINE_INTERVAL_1MIN},
      {"3m", fastfeed::proto::KLINE_INTERVAL_3MIN},
      {"5m", fastfeed::proto::KLINE_INTERVAL_5MIN},
      {"15m", fastfeed::proto::KLINE_INTERVAL_15MIN},
      {"30m", fastfeed::proto::KLINE_INTERVAL_30MIN},
      {"1h", fastfeed::proto::KLINE_INTERVAL_1HOUR},
      {"1d", fastfeed::proto::KLINE_INTERVAL_1DAY}
    };

    auto [iter, succ] = interval_map.emplace(interval, fastfeed::proto::KLINE_INTERVAL_INVALID);
    return iter->second;
  }

  static fastfeed::proto::TradingDataInterval ParseTradingDataInterval(std::string_view interval) {
    static std::map<std::string, fastfeed::proto::TradingDataInterval> interval_map = {
      {"5m", fastfeed::proto::TRADING_DATA_INTERVAL_5MIN},
      {"15m", fastfeed::proto::TRADING_DATA_INTERVAL_15MIN},
      {"1h", fastfeed::proto::TRADING_DATA_INTERVAL_1HOUR},
      {"1d", fastfeed::proto::TRADING_DATA_INTERVAL_1DAY}
    };

    auto [iter, succ] = interval_map.emplace(interval, fastfeed::proto::TRADING_DATA_INTERVAL_INVALID);
    return iter->second;
  }

  template<class T>
  static bool ParseKlineWs(
      T&& msg,
      Klines* klines,
      fastfeed::proto::KlineInterval interval) {
    CHECK(klines);
    auto v = msg["k"];
    int64_t open_timestamp = int64_t(v["t"]) * 1'000'000LL;
    auto result = klines->AddKline(open_timestamp);
    if (!result) return false;

    Kline* kline;
    bool freezed;
    std::tie(kline, freezed) = result.value();
    kline->timestamp = int64_t(msg["E"]) * 1'000'000LL;
    kline->open_timestamp = open_timestamp;
    kline->open = double(v["o"].get_double_in_string());
    kline->high = double(v["h"].get_double_in_string());
    kline->low = double(v["l"].get_double_in_string());
    kline->close = double(v["c"].get_double_in_string());
    kline->close_timestamp = int64_t(v["T"]) * 1'000'000LL;
    kline->volume = double(v["v"].get_double_in_string());
    kline->interval = interval;

    return freezed;
  }

  template<class T>
  static bool ParseTicker(
      T&&  msg,
      Klines* klines) {
    CHECK(klines);
    auto* kline = klines->AddKline();
    kline->timestamp = int64_t(msg["E"]) * 1'000'000LL;
    kline->open_timestamp = kline->timestamp - 24 * 60 * 60 * 1'000'000'000LL;
    kline->open = double(msg["o"].get_double_in_string());
    kline->high = double(msg["h"].get_double_in_string());
    kline->low = double(msg["l"].get_double_in_string());
    kline->close = double(msg["c"].get_double_in_string());
    kline->close_timestamp = kline->timestamp;
    kline->volume = double(msg["v"].get_double_in_string());
    kline->turnover = double(msg["q"].get_double_in_string());
    kline->interval = fastfeed::proto::KLINE_INTERVAL_ROLLING_1DAY;

    return true;
  }

  template<class T>
  static bool ParseKlineRest(
      T&& v,
      Klines* klines,
      fastfeed::proto::KlineInterval interval,
      int64_t timestamp
      ) {
    // CHECK_EQ(v.count_elements(), 12);

    int i = 0;
    Kline* kline = nullptr;
    bool freezed = true;
    for (auto &&val : v) {
      switch(i++) {
        case 0:
        {
          int64_t open_timestamp = int64_t(val) * 1'000'000LL;
          auto result = klines->AddKline(open_timestamp);
          if (!result) return false;
          std::tie(kline, freezed) = result.value();
          kline->open_timestamp = open_timestamp;
          kline->timestamp = timestamp;
          kline->interval = interval;
          break;
        }
        case 1:
          kline->open = double(val.get_double_in_string());
          break;
        case 2:
          kline->high = double(val.get_double_in_string());
          break;
        case 3:
          kline->low = double(val.get_double_in_string());
          break;
        case 4:
          kline->close = double(val.get_double_in_string());
          break;
        case 5:
          kline->volume = double(val.get_double_in_string());
          break;
        case 6:
          kline->close_timestamp = int64_t(val) * 1'000'000LL;
          break;
        case 7:
          kline->turnover = double(val.get_double_in_string());
          break;
        case 9:
          kline->buy_volume = double(val.get_double_in_string());
          break;
        case 10:
          kline->buy_turnover = double(val.get_double_in_string());
          break;
        default:
          break;
      }
    }
    return freezed;
  }

  template<class T>
  static void ParseWallet(
      T&&  msg,
      Wallet* wallet) {
    wallet->can_deposit = bool(msg["depositAllEnable"]);
    wallet->can_withdraw = bool(msg["withdrawAllEnable"]);
  }

  template<typename T>
  static bool ParseAccountRatio(
      simdjson::ondemand::object& msg,
      TradingData<T>* trading_data) {
    int64_t ts = int64_t(msg["timestamp"]) * 1'000'000LL;
    auto first = trading_data->Get();
    if (first) {
      // all are freezed
      if (first.value()->timestamp >= ts) return false;
    }
    if (!isdigit(msg["longShortRatio"].get_string().value()[0])) return false;

    T* ratio = trading_data->Add();
    ratio->timestamp = ts;
    ratio->long_account = stod(msg["longAccount"].get_string());
    ratio->long_short_ratio = stod(msg["longShortRatio"].get_string());
    ratio->short_account = stod(msg["shortAccount"].get_string());
    return true;
  }

  static bool ParsePositionRatio(
      simdjson::ondemand::object& msg,
      TradingData<TopLongShortPositionRatio>* trading_data) {
    int64_t ts = int64_t(msg["timestamp"]) * 1'000'000LL;
    auto first = trading_data->Get();
    if (first) {
      // all are freezed
      if (first.value()->timestamp >= ts) return false;
    }
    if (!isdigit(msg["longShortRatio"].get_string().value()[0])) return false;

    TopLongShortPositionRatio* ratio = trading_data->Add();
    ratio->timestamp = ts;
    ratio->long_position= stod(msg["longAccount"].get_string());
    ratio->long_short_ratio = stod(msg["longShortRatio"].get_string());
    ratio->short_position = stod(msg["shortAccount"].get_string());
    return true;
  }

  static bool ParseTakerRatio(
      simdjson::ondemand::object& msg,
      TradingData<TakerBuySellRatio>* trading_data) {
    int64_t ts = int64_t(msg["timestamp"]) * 1'000'000LL;
    auto first = trading_data->Get();
    if (first) {
      // all are freezed
      if (first.value()->timestamp >= ts) return false;
    }

    TakerBuySellRatio* ratio = trading_data->Add();
    ratio->timestamp = ts;
    ratio->buy_sell_ratio = stod(msg["buySellRatio"].get_string());
    ratio->buy_volume = stod(msg["buyVol"].get_string());
    ratio->sell_volume = stod(msg["sellVol"].get_string());
    return true;
  }

  static bool ParseOpenInterestHist(
      simdjson::ondemand::object& msg,
      TradingData<OpenInterest>* trading_data) {
    int64_t ts = int64_t(msg["timestamp"]) * 1'000'000LL;
    auto first = trading_data->Get();
    if (first) {
      // all are freezed
      if (first.value()->timestamp >= ts) return false;
    }

    OpenInterest* open_interest = trading_data->Add();
    open_interest->timestamp = ts;
    open_interest->open_interest_qty = stod(msg["sumOpenInterest"].get_string());

    return true;
  }
};

}  // namespace impl

using impl::BinanceBaseFeedUpdateParser;

}  // namespace coin2::exchange::binance_base::feed_v1
